package com.wjw.redis.stream.handler;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.wjw.redis.stream.enums.RedisDelayQueueEnum;
import com.wjw.redis.stream.util.RedisDelayQueueUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author wjwei
 */
@Slf4j
@Component
public class RedisDelayQueueRunner implements CommandLineRunner {
    @Autowired
    private RedisDelayQueueUtil redisDelayQueueUtil;
    @Autowired
    private ApplicationContext context;
    @Resource(name = "TaskExecutor")
    private ThreadPoolTaskExecutor poolTaskExecutor;

    ThreadPoolExecutor executorService = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS,
        new LinkedBlockingQueue<Runnable>(1000), new ThreadFactoryBuilder().setNameFormat("order-delay-%d").build());

    @Override
    public void run(String... args) throws Exception {
        poolTaskExecutor.execute(() -> {
            boolean interrupted = false;
            while (!interrupted) {
                try {
                    RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
                    for (RedisDelayQueueEnum queueEnum : queueEnums) {
                        try {
                            Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
                            if (value != null) {
                                @SuppressWarnings("unchecked")
                                RedisDelayQueueHandle<Object> redisDelayQueueHandle = (RedisDelayQueueHandle<Object>) context.getBean(queueEnum.getBeanId());
                                executorService.execute(() -> {
                                    try {
                                        redisDelayQueueHandle.execute(value);
                                    } catch (Exception e) {
                                        log.error("(Redisson延迟队列处理异常) {}, Queue: {}, Value: {}", e.getMessage(), queueEnum, value);
                                    }
                                });
                            }
                        } catch (Exception e) {
                            log.error("(Redisson延迟队列获取异常) {}, Queue: {}", e.getMessage(), queueEnum);
                        }
                    }
                    try {
                        TimeUnit.MILLISECONDS.sleep(500);
                    } catch (InterruptedException e) {
                        log.error("(Redisson延迟队列监测异常中断) {}", e.getMessage());
                        interrupted = true; // 标记中断，退出循环
                        Thread.currentThread().interrupt(); // 恢复中断状态
                    }
                } catch (Exception e) {
                    log.error("(Redisson延迟队列监测异常中断) ", e);
                }
            }
        });
        log.info("(Redisson延迟队列监测启动成功)");
    }
}
